iT邦幫忙

2025 iThome 鐵人賽

DAY 9
0
自我挑戰組

30 天工程師雜學之旅系列 第 9

MCP-3 用 FastAPI 打造 NASA API Baseline(為 MCP 整合做準備)

  • 分享至 

  • xImage
  •  

你會得到什麼

  • 一個可直接啟動的 FastAPI 專案
  • 兩支可用 API:
    • GET /apod:回傳每日天文圖(Astronomy Picture of the Day)
    • GET /mars-photo:回傳火星照片
    • GET /neo:回傳近地天體(Near-Earth Objects)資訊
  • 完整檔案與步驟,下一篇可無痛接上 MCP 工具化

0) 需求與安裝

系統需求

專案結構

此為簡單的 FastAPI 專案,正式的作業環境應該把 route 分離到不同的檔案中。此專案結構僅作demo用:

nasa-app-mcp-demo/
├── .env
├── README.md
├── main.py
├── test_basic.py
├── venv/
│   └── ... (virtual environment files)

requirements.txt

fastapi>=0.104.0
uvicorn[standard]>=0.24.0
httpx>=0.25.0
python-dotenv>=1.0.0
fastapi-mcp>=0.4.0

1) 建立環境

# 建資料夾
mkdir -p nasa-fastapi-mcp-demo/app && cd nasa-fastapi-mcp-demo

# 建立虛擬環境
python3.12 -m venv .venv && source .venv/bin/activate

# 建 requirements.txt(貼上上面內容)
# 然後安裝
pip install -r requirements.txt

2) 設定環境變數

建立 .env.example

NASA_API_KEY=your_nasa_api_key_here

複製一份:

cp .env.example .env
# 然後把 your_nasa_api_key_here 改成你自己的 Key

3) 加入程式碼

main.py

import os
import httpx
from typing import Optional, List, Dict, Any
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

app = FastAPI(
    title="Simple NASA MCP Server",
    description="A simple FastAPI server providing NASA data through REST endpoints and MCP",
    version="1.0.0"
)

# NASA API configuration
NASA_API_KEY = os.getenv("NASA_API_KEY", "DEMO_KEY")
NASA_BASE_URL = "https://api.nasa.gov"

# HTTP client for NASA API calls
client = httpx.AsyncClient(timeout=30.0)

# Response models
class APODResponse(BaseModel):
    title: str
    explanation: str
    url: str
    date: str
    media_type: str
    hdurl: Optional[str] = None
    copyright: Optional[str] = None

class MarsPhotoResponse(BaseModel):
    id: int
    sol: int
    camera: Dict[str, Any]
    img_src: str
    earth_date: str
    rover: Dict[str, Any]

class MarsPhotosResponse(BaseModel):
    photos: List[MarsPhotoResponse]

class NEOResponse(BaseModel):
    id: str
    name: str
    estimated_diameter: Dict[str, Any]
    is_potentially_hazardous_asteroid: bool
    close_approach_data: List[Dict[str, Any]]

class NEOFeedResponse(BaseModel):
    element_count: int
    near_earth_objects: Dict[str, List[NEOResponse]]



@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "service": "simple-nasa-mcp"}

@app.get("/apod", response_model=APODResponse)
async def get_astronomy_picture_of_day(date: Optional[str] = Query(None, description="Date in YYYY-MM-DD format")):
    """Get NASA's Astronomy Picture of the Day"""
    try:
        params = {"api_key": NASA_API_KEY}
        if date:
            params["date"] = date
            
        response = await client.get(f"{NASA_BASE_URL}/planetary/apod", params=params)
        
        if response.status_code == 200:
            data = response.json()
            return APODResponse(**data)
        else:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"NASA API error: {response.text}"
            )
    except httpx.RequestError as e:
        raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@app.get("/mars-photos", response_model=MarsPhotosResponse)
async def get_mars_rover_photos(
    rover: str = Query(..., description="Rover name (curiosity, perseverance, opportunity, spirit)"),
    sol: int = Query(..., description="Martian sol (day) number"),
    camera: Optional[str] = Query(None, description="Camera name (FHAZ, RHAZ, MAST, NAVCAM, etc.)")
):
    """Get Mars rover photos"""
    try:
        params = {
            "api_key": NASA_API_KEY,
            "sol": sol
        }
        if camera:
            params["camera"] = camera
            
        response = await client.get(f"{NASA_BASE_URL}/mars-photos/api/v1/rovers/{rover}/photos", params=params)
        
        if response.status_code == 200:
            data = response.json()
            return MarsPhotosResponse(**data)
        else:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"NASA API error: {response.text}"
            )
    except httpx.RequestError as e:
        raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@app.get("/neo", response_model=NEOFeedResponse)
async def get_near_earth_objects(
    start_date: str = Query(..., description="Start date in YYYY-MM-DD format"),
    end_date: str = Query(..., description="End date in YYYY-MM-DD format")
):
    """Get Near Earth Objects (asteroids and comets)"""
    try:
        params = {
            "api_key": NASA_API_KEY,
            "start_date": start_date,
            "end_date": end_date
        }
        
        response = await client.get(f"{NASA_BASE_URL}/neo/rest/v1/feed", params=params)
        
        if response.status_code == 200:
            data = response.json()
            return NEOFeedResponse(**data)
        else:
            raise HTTPException(
                status_code=response.status_code,
                detail=f"NASA API error: {response.text}"
            )
    except httpx.RequestError as e:
        raise HTTPException(status_code=503, detail=f"Failed to connect to NASA API: {str(e)}")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")

@app.on_event("shutdown")
async def shutdown_event():
    """Clean up HTTP client on shutdown"""
    await client.aclose()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

4) 跑起來!

uvicorn main:app --reload
  • --reload 參數會在代碼變更時自動重啟伺服器
    或是直接執行
python main.py

這會啟動 FastAPI 伺服器,並在 http://127.0.0.1:8000 提供 API。

  • Swagger UI:http://127.0.0.1:8000/docs
    swaggerui

5) 測試 API

# 健康檢查
curl "http://127.0.0.1:8000/health"

# 今日 APOD
curl "http://127.0.0.1:8000/apod"

# 指定日期 APOD
curl "http://127.0.0.1:8000/apod?date=2024-07-20&hd=true"

# 搜尋圖片
curl "http://127.0.0.1:8000/mars-photos?rover=curiosity&sol=1000&camera=FHAZ"

# 近地天體
curl "http://127.0.0.1:8000/neo?start_date=2024-07-01&end_date=2024-07-31"

6) 常見問題

  • 503:NASA API 服務不可用 → 請稍後再試 或是換一個api試試看
  • 401 / 403:API Key 未設定或錯誤 → 確認 .env
  • 429:API 流量限制 → 換關鍵字或加快取
  • 日期格式:請用 YYYY-MM-DD

下一篇預告:把 REST 變成 AI 工具

下一篇我們會用 fastapi-mcp 直接將這些endpoint轉為MCP可發現的tools,讓 VS Code 可以直接 discover 與呼叫。


上一篇
MCP-2 MCP 原理解析 — AI 如何「發現」與「使用」你的工具
下一篇
MCP-4 在 FastAPI 中整合 MCP 並暴露工具
系列文
30 天工程師雜學之旅23
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言